package com.jugheadzhou.core.zk.helper;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Maps;
import com.jugheadzhou.core.zk.ZkClient;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.framework.recipes.leader.LeaderLatchListener;

import java.util.Map;

@Slf4j
public class MythServiceLeaderLatchHelper extends MythServiceBaseZkHelper {

    private LeaderLatch leaderLatch;

    private static String LEADER = "/leader";
    private static ObjectMapper objectMapper = new ObjectMapper();

    private String ledaerPathUnderServiceNode;

    public MythServiceLeaderLatchHelper(ZkClient zkClient, String ip, String port) {
    	 super(zkClient, ip, port);
    }

    public MythServiceLeaderLatchHelper(ZkClient zkClient, String ip, String port, String ledaerPathUnderServiceNode) {
        super(zkClient, ip, port);
        this.ledaerPathUnderServiceNode = ledaerPathUnderServiceNode;
    }

    public void start(LeaderLatchListener leaderLatchListener) throws Exception {
        String servicePath = buildServicePath();
        if (ledaerPathUnderServiceNode == null || ledaerPathUnderServiceNode.trim().length() == 0) {
            ledaerPathUnderServiceNode = LEADER;
        } else if (!ledaerPathUnderServiceNode.startsWith(ZkClient.SLASH)) {
            ledaerPathUnderServiceNode = ZkClient.SLASH + ledaerPathUnderServiceNode;
        }
        String leaderPath = servicePath + ledaerPathUnderServiceNode;
        Map<String, String> metaData = Maps.newHashMap();
        metaData.put("server", this.getServiceLocalUrl());
        String id = objectMapper.writeValueAsString(metaData);
        log.info("start leader latch: " + leaderPath + " with id[" + id + "]");

        leaderLatch = new LeaderLatch(zkClient.getClient(), leaderPath, id);
        if (leaderLatchListener != null)
            leaderLatch.addListener(leaderLatchListener);
        leaderLatch.start();
    }

    public void stop() throws Exception {
        this.leaderLatch.close();
    }

    public boolean isLeader() {
        return this.leaderLatch.hasLeadership();
    }
}
